package org.marsdonne.sync.messenger.artemis;

import org.marsdonne.sync.Event;
import org.marsdonne.sync.Synced;
import org.marsdonne.sync.SyncedImpl;
import org.springframework.context.ApplicationContext;
import org.springframework.jms.annotation.JmsListener;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class SyncedImplListener {

    private final Map<String, List<Synced>> syncedRegistry = new LinkedHashMap<>();
    private final int maximumPoolSize = (int) Math.ceil(Runtime.getRuntime().availableProcessors());
    private final ExecutorService executor = new ThreadPoolExecutor(
            1, maximumPoolSize, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(maximumPoolSize * 10));

    @Resource
    private ApplicationContext context;

    public void scanSyncedImpl() {
        Set<Map.Entry<String, Object>> syncedSet = context.getBeansWithAnnotation(SyncedImpl.class).entrySet();

        for (Map.Entry<String, Object> beanEntry : syncedSet) {
            if (beanEntry.getValue() instanceof Synced) {
                SyncedImpl syncedImpl = context.findAnnotationOnBean(beanEntry.getKey(), SyncedImpl.class);
                registerSynchronizer(syncedImpl.domain(), (Synced) beanEntry.getValue());
            }
        }
    }

    public void registerSynchronizer(String domain, Synced synchronizer) {
        List<Synced> syncedList = syncedRegistry.getOrDefault(domain, new ArrayList<>());
        syncedList.add(synchronizer);
        syncedRegistry.put(domain, syncedList);
    }

    @JmsListener(
            containerFactory = "syncedImplListenerFactory", concurrency = "1-5",
            destination = "${sync.artemis.channel:channels.sync}")
    public void receive(Event event) {
        //TODO:内置幂等机制，避免在集群多节点环境下，因“组播”造成的“数据同步风暴”（虽然针对副本数据的重复更新不存在幂等性问题，或具体的同步器会解决基于业务的幂等问题，但仍会造成不必要的存储I/O开销）
        List<Synced> synchronizes = syncedRegistry.get(event.getDomain());

        if (synchronizes != null && synchronizes.size() > 0) {
            for (Synced synchronizer : synchronizes) {
                executor.execute(() -> synchronizer.sync(event));
            }
        }
    }
}
